package com.liao.dao;


import com.liao.utils.ConnectionInstance;
import com.liao.utils.HbaseUtil;
import com.liao.utils.PropertyUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.List;

/**
 * 创建namespace，创建表，put数据
 *
 */
public class HbaseDAO {

    private String namespace;
    private String tableName;
    private int regions;//HBase分区数
    private String cf;//HBase列族
    private SimpleDateFormat sdf=null;
    private HTable table;//操作HBase表对象
    private String flag;//主被叫数据标志位
    //缓存put对象集合
    private List<Put> listPut;
    private String cf2;
    //在创建HBaseDAO时初始化属性及创建NS和table
    public HbaseDAO() throws IOException {
        //初始化相关属性（数据来源于配置文件kafka_hbase.properties）
        namespace= PropertyUtil.properties.getProperty("hbase.namespace");
        tableName=PropertyUtil.properties.getProperty("hbase.table.name");
        regions=Integer.valueOf(PropertyUtil.properties.getProperty("hbase.regions"));
        cf=PropertyUtil.properties.getProperty("hbase.table.cf");
        cf2 = PropertyUtil.properties.getProperty("hbase.table.cf2");
        sdf=new SimpleDateFormat("yyy-MM-dd HH:mm:ss");
        listPut=new ArrayList<>();
        flag="1";
        //初始化命名空间及表的创建
        if(!HbaseUtil.isTableExist(tableName)){
            HbaseUtil.initNamespace(namespace);
            HbaseUtil.createTable(tableName,regions,cf,cf2);
        }
    }
    //ori原始数据
    public void put(String ori) throws Exception {
        //创建连接及获取表
        //通过工厂模式获取连接
        if(listPut.size()==0){
            //获取连接（单例对象）
            //Connection connection = ConnectionFactory.createConnection(Constant.CONF);
            Connection connection= ConnectionInstance.getInstance();
            //获取表对象
            table=(HTable)connection.getTable(TableName.valueOf(tableName));
            //设置不自动提交
            table.setAutoFlushTo(false);
            //设置客户端缓存的大小
            table.setWriteBufferSize(1024*1024);
        }

        //如果传输的数据为空直接返回
        if (ori==null) return;
        String[] split = ori.split( ",");//切分数据
        String caller=split[0];//主叫
        String callee=split[1];//被叫
        String buildTime=split[2];//建立通话时间
        long time=sdf.parse(buildTime).getTime();//通话时间时间戳
        //long类型转化为string类型
        String bulidtime_ts=time+"";//时间戳转换位String类型
        String duration=split[3];//获取通话时长
        //获得分区号
        String regionHash=HbaseUtil.getRegionHash(caller,buildTime,regions);
        //获取rowkey
        String rowKey=HbaseUtil.getRowKey(regionHash,caller,buildTime,callee,flag,duration);
        Put put = new Put(Bytes.toBytes(rowKey));
        put.addColumn(Bytes.toBytes(cf),Bytes.toBytes("call1"),Bytes.toBytes(caller));
        put.addColumn(Bytes.toBytes(cf),Bytes.toBytes("builTime"),Bytes.toBytes(buildTime));
        put.addColumn(Bytes.toBytes(cf),Bytes.toBytes("bulidtime_ts"),Bytes.toBytes(bulidtime_ts));
        put.addColumn(Bytes.toBytes(cf),Bytes.toBytes("call2"),Bytes.toBytes(callee));
        put.addColumn(Bytes.toBytes(cf),Bytes.toBytes("flag"),Bytes.toBytes(flag));
        put.addColumn(Bytes.toBytes(cf),Bytes.toBytes("duration"),Bytes.toBytes(duration));
        //向put缓存中添加对象
        listPut.add(put);
        //当list中数据条数到达20条，则写入hbase
        if(listPut .size()>20){
            table.put(listPut);
            //手动提交
            table.flushCommits();
            //清空list集合
            listPut.clear();
            //关闭表连接（如果业务单一，可以初始化时创建表连接，此处就不需要关闭表连接）
            table.close();
        }
    }
}
